package wordcount;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.StringTokenizer;

import barrier.Barrier;

import com.ibm.tspaces.Field;
import com.ibm.tspaces.Tuple;
import com.ibm.tspaces.TupleSpace;
import com.ibm.tspaces.TupleSpaceException;

public class Slave {

  static final String PART = "partition";
  static final String DONE = "done";

  static TupleSpace ts;
  static TupleSpace bs;
  /*Se usarmos multiplos Java threads, 
   * este map terá que ser um concurrent hash map*/
  static Map<String, Integer> map;
  static Tuple tuple;



  private static String getPartition() 
  throws TupleSpaceException
  {
    tuple = new Tuple(PART, new Field(String.class));
    Tuple t = ts.take(tuple);

    if(t != null)
      return (String) t.getField(1).getValue();

    return null;
  }

  /* Se quisermos paralelizar o slave, 
   * este work() terá que um runnable no qual cada thread irá executar.
   * O parametro de entrada será uma parte da partição obtida.*/
  private static void work(String text) 
  throws TupleSpaceException
  {
    StringTokenizer st = new StringTokenizer(text);

    while(st.hasMoreElements())
    {
      String s1 = st.nextToken();

      if(map.containsKey(s1))
      {
        int n = map.get(s1);
        map.put(s1, n + 1);	
      }
      else
        map.put(s1, 1);
    }
  }

  private static void done() 
  throws TupleSpaceException
  {
    //Enviar resultados para o espaço de tuplos

    Iterator<Entry<String, Integer>> it = map.entrySet().iterator();
    while(it.hasNext())
    {
      Entry<String, Integer> e = it.next();
      String word = e.getKey();
      int count = e.getValue();

      tuple = new Tuple(word, count);
      ts.write(tuple);
    }

    //Não existem mais partições.
    tuple = new Tuple(DONE);
    ts.write(tuple);
    
    System.out.println("Slave acabou.");
  }

  public static void main(String args[])
  {
    /*
     * args[0] => Número de slaves
     * args[1] => IP do master
     */

    /*
     * 1) Enquanto houverem partições:
     * 	  a) Obtém partição - Uma string;
     * 	  b) Obtém o conjunto de palavras presentes no texto;
     *    c) Para cada palavra, incrementar o seu numero de ocorrências
     * 
     * 2) Se não houverem partições:
     * 	  a) Coloca tuplo <done> quando acabar.
     */

    try {
      ts = new TupleSpace("wordCount", args[1]);
      bs = new TupleSpace("barrier", args[1]);
      
      Barrier barrier = new Barrier("barrier", bs);
      System.err.print("À espera... ");
      barrier.barrier(Integer.parseInt(args[0])+1);
      System.err.println("OK!");
      
      map = new HashMap<String, Integer>();

      String partition;

      while( (partition = getPartition()) != null)
        work(partition);

      done();

    }catch (TupleSpaceException e) {e.printStackTrace();}
  }

}
